package com.tc.vms.producer;

import com.tc.vms.MessageWrap;
import com.tc.vms.SLA;
import com.tc.vms.connection.VmsConnectionManager;
import com.tc.vms.exception.VmsClientErrorCode;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.iface.*;
import com.tc.vms.message.MessageConst;
import com.tc.vms.message.SendResultType;
import com.tc.vms.utils.Utils;
import com.yammer.metrics.core.TimerContext;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * Created by yonghua.zhang on 2016/1/7.
 */
public class VmsProducerProxy extends VmsConnectionWrap implements IPublisher {
    private volatile IPublisher publisher;
    private final VmsProducerMetrics vmsProducerMetrics;
//    private final VmsHermesManager.VmsHermesProducerStat vmsHermesProducerStat;

    public VmsProducerProxy(VmsConnectionManager vmsConnectionManager, VmsMetaData vmsMetaData, SLA serviceSLA,
                            VmsQos vmsQos, Properties props) {
        super(vmsConnectionManager, vmsMetaData, serviceSLA, vmsQos, props);
        vmsProducerMetrics = getProducerMetrics();
//        vmsHermesProducerStat = new VmsHermesManager.VmsHermesProducerStat(vmsMetaData.getOrigClientId(),
//                Utils.getHostIp(), vmsMetaData.getZnodeName(), vmsProducerMetrics);
    }

    @Override
    protected double getRealTpsInOneMinute() {
        return Math.max(vmsProducerMetrics.messageSuccessRate().oneMinuteRate(),
                vmsProducerMetrics.messageFailureRate().oneMinuteRate());
    }

    public VmsProducerProxy(VmsConnectionManager vmsConnectionManager, VmsMetaData vmsMetaData, Properties props)
            throws VmsClientException {
        this(vmsConnectionManager, vmsMetaData, new SLA(), new PubQoS(), props);
    }

    @Override
    public void suspend() {
        this.isSuspend = true;
    }

    @Override
    public void resume() {
        this.isSuspend = false;
    }

    /**
     * for outer user call, inner invoke need using sla() or member directly
     *
     * @return
     */
    public final SLA slaOption() {
        try {
            chkResetOption();
            return super.sla();
        } catch (VmsClientException e) {
            LOGGER.warn("Vmsclient is running,couldn't modify service option.", e);
        }
        return super.sla().cloneSLA(); // return new option to avoid NPE
    }

/*    public final PubQoS qosOption() {
        try {
            chkResetOption();
            return super.qos();
        } catch (VmsClientException e) {
            LOGGER.warn("Vmsclient is running,couldn't modify service option.", e);
        }
        return new PubQoS(); // return new option to avoid NPE
    }*/

    SLA getSLA() {
        return this.serviceSLA;
    }

    PubQoS getQos() {
        return (PubQoS) this.vmsQos;
    }

    public synchronized void zkNotifyReconnectServer(VmsMetaData newMetaData, VmsMetaData origMetaData) {
        this.vmsConnectionManager.updateConnectionWrap(this, origMetaData, newMetaData);
        if (publisher != null) { //close inner publisher service
            publisher.shutdown();
            publisher = null;
        }
        int retryTimes = 0;
        boolean rebootOk = false;
        while (isOpen && retryTimes++ < 3) {
            LOGGER.debug("VmsProducerProxy zkNotifyReconnectServer restart service for [{}] times.", retryTimes);
            try {
                init(null);
                if (publisher != null) {
                    publisher.start();
                } else {
                    throw new VmsClientException("zkNotifyReconnectServer Init publisher failed.");
                }
                rebootOk = true;
                LOGGER.debug("VmsProducerProxy zkNotifyReconnectServer restart service successfully.");
                break;
            } catch (VmsClientException e) {
                LOGGER.warn("ReConnectServer error, retry after 3s interval.", e);
                Utils.sleep3s();
                if (publisher != null) {
                    publisher.shutdown();
                    publisher = null;
                }
            }
        }

        if (!rebootOk) {
            LOGGER.error("zkNotifyReconnectServer service couldn't reboot. please check the config [{}].", newMetaData);
        }
    }

    @Override
    public synchronized boolean init(IContext context) throws VmsClientException {
        LOGGER.debug("VmsProducerProxy start initializing... ");
        if (publisher == null) {
            ChannelMetaData channelMetaData = getVmsMetaData();
            switch (channelMetaData.getBrokerType()) {
                case KAFKA:
                    LOGGER.debug("VmsProducerProxy initialize kafka producer instance. ");
                    publisher = new VmsProducerKafkaImpl(this, vmsConnectionManager);
                    break;
                case RABBITMQ:
                    LOGGER.debug("VmsProducerProxy initialize rabbitmq producer instance. ");
                    publisher = new VmsProducerRmqMultiQueueImpl(this, vmsConnectionManager);
                    break;
                default:
                    throw new VmsClientException(
                            "ProducerFactory don't support publisher type [" + channelMetaData.getBrokerType() + "].",
                            VmsClientErrorCode.VMS_SYS_ERR_PRODUCER_TYPE_NOT_SUPPORT);
            }
            publisher.init(context);
        }
        return true;
    }

    @Override
    public synchronized boolean start() throws VmsClientException {
        if (!isOpen) {
            init(null);
            publisher.start();
            registerZkNode();
//            vmsHermesProducerStat.registerHermes();
            isOpen = true;
            LOGGER.info("VmsProducerProxy [{}] start successfully.", vmsMetaData.getClientInstName());
        }
        return isOpen;
    }

    @Override
    public void shutdown() {
        if (isOpen) {
            isOpen = false;
            super.shutdown();
            if (publisher != null) {
                try {
                    publisher.shutdown();
                    vmsConnectionManager.getVmsMetricsManager().removeVmsMetrics(vmsMetaData);
//                    vmsHermesProducerStat.unRegisterHermes();
                } catch (Exception e) {
                }
                publisher = null;
                LOGGER.info("VmsProducerProxy [{}] shutdown successfully.", vmsMetaData.getClientInstName());
            }
        }
    }

    private void addMessageExtAttrs(MessageWrap message) {
        try {
            if (message.getMessageId() == null || message.getMessageId().isEmpty()) {
                String msgId = msgGenerator.genMessageId().encode();
                message.setMessageId(msgId);
            }
            String bornHost = (String) message.getAttribute(MessageConst.PROPERTY_BORN_HOST);
            if (bornHost == null || bornHost.isEmpty()) {
                message.setAttribute(MessageConst.PROPERTY_BORN_HOST, MessageConst.HOSTNAME);
            }
            /*message.setAppId(vmsMetaData.getBelongAttr().getBelongAppname());
            String userId = vmsMetaData.getUserName();
			if (userId != null && !userId.isEmpty()) {
				message.setUserId(userId);
			}*/
        } catch (Exception e) {
            LOGGER.warn("Add message info failed.", e);
        }
    }

    @Override
    public SendResult sendMessage(MessageWrap message) throws VmsClientException {
        return sendMessageByOrder(message, null);
    }

    @Override
    public SendResult sendMessageByOrder(MessageWrap message, String partKey) throws VmsClientException {
        SendResult sr = null;

        try {

            chkMessage(message);
            chkServiceOk();

            long startTime = System.currentTimeMillis();
            sr = publisher.sendMessageByOrder(message, partKey);
            vmsProducerMetrics.sendCostRate().update(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS);

            if (sr.getSendResultType().equals(SendResultType.VMS_SEND_OK)) {
                vmsProducerMetrics.messageSuccessRate().mark();
                vmsProducerMetrics.byteRate().mark(message.getPayload().length);
            } else {
                vmsProducerMetrics.messageFailureRate().mark();
            }

        } catch (VmsClientException e) {
            if (e.getErrorCode() == VmsClientErrorCode.VMS_SYS_RUNTIME_ERR_SUSPEND_SERVICE) {
                LOGGER.debug("Service is suspend by vms center.");
                sr = new SendResult(SendResultType.VMS_SEND_SUSPEND, message.getMessageId(), e.getErrMsg());

            } else {
                LOGGER.debug("sendMessageByOrder error.", e);
                throw e;
            }
        } catch (Exception ex) {
            LOGGER.debug("sendMessageByOrder error.", ex);
            sr = new SendResult(SendResultType.VMS_SEND_FAILED, message.getMessageId(), ex.getMessage());

        }
        LOGGER.debug("VmsProducerProxy send message = [{}] status = [{}]", message.getMessageId(),
                sr.getSendResultType());
        return sr;
    }

    @Override
    public SendResult sendTransactionMessage(MessageWrap message, TransactionMessageListener listener)
            throws VmsClientException {
        SendResult sr = null;

        try {

            chkMessage(message);
            chkServiceOk();

            TimerContext tc = vmsProducerMetrics.sendCostRate().time();
            sr = publisher.sendTransactionMessage(message, listener);
            tc.stop();
            if (sr.getSendResultType().equals(SendResultType.VMS_SEND_OK)) {
                vmsProducerMetrics.messageSuccessRate().mark();
                vmsProducerMetrics.byteRate().mark(message.getPayload().length);
            } else {
                vmsProducerMetrics.messageFailureRate().mark();
            }

        } catch (VmsClientException e) {
            if (e.getErrorCode() == VmsClientErrorCode.VMS_SYS_RUNTIME_ERR_SUSPEND_SERVICE) {
                LOGGER.debug("Service is suspend by vms center.");
                sr = new SendResult(SendResultType.VMS_SEND_FAILED, message.getMessageId(), e.getErrMsg());
            } else {
                throw e;
            }
        }
        LOGGER.debug("VmsProducerProxy send message = [{}] status = [{}]", message.getMessageId(),
                sr.getSendResultType());
        return sr;
    }

    private void chkMessage(MessageWrap messageWrap) throws VmsClientException {
        if (messageWrap == null || messageWrap.getPayload() == null) {
            throw new VmsClientException("Messagewrap is null, or Message body is null");
        }
        if (messageWrap.getPayload().length > sla().messageMaxSize()) {
            throw new VmsClientException(String.format("The message size [%d] greater than size limit [%d].",
                    messageWrap.getPayload().length, sla().messageMaxSize()));
        }
        addMessageExtAttrs(messageWrap);
    }

    private VmsProducerMetrics getProducerMetrics() {
        return this.vmsConnectionManager.getVmsMetricsManager().getVmsMetrics(this.vmsMetaData);
    }

}
